package ex.datastream;

import ex.datastream.functions.richFunction.RichFlatMapFunc;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

/**
输入
1,3
2,4
1,5
1,7
2,6
输出（按分组累加）
 (1,3)
 (1,4)
 (2,8)
 (3,15)
 (2,10)
 */
public class CountWindowAverageJob extends ApiFrame {
    public static void main(String[] args) throws Exception {
        CountWindowAverageJob job = new CountWindowAverageJob();
        job.getEnv();

        DataStream<Tuple2<Long, Long>> dataStream = job.env.fromElements(
                Tuple2.of(1L, 3L), Tuple2.of(2L, 4L), Tuple2.of(1L, 5L), Tuple2.of(1L, 7L), Tuple2.of(2L, 6L));

        KeyedStream<Tuple2<Long, Long>, Long> keyedStream = dataStream.keyBy(value -> value.f0);
        SingleOutputStreamOperator<Tuple2<Long, Long>> operator = keyedStream.flatMap(new RichFlatMapFunc());
        operator.printToErr();
        job.env.execute("CountWindowAverageJob");
    }


}
